package sandan.kafka.productor;

import java.util.Properties;
import java.util.Random;

import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import sandan.tools.DateFmt;
import backtype.storm.utils.Utils;

public class Producer extends Thread {

	private final kafka.javaapi.producer.Producer<Integer, String> producer;
	private final String topic;
	private final Properties props = new Properties();

	public Producer(String topic) {
		props.put("serializer.class", "kafka.serializer.StringEncoder"); // 字符串消息
		props.put("metadata.broker.list", KafkaProperties.brokerList);
		producer = new kafka.javaapi.producer.Producer<Integer, String>(
				new ProducerConfig(props));
		this.topic = topic;
	}

	@Override
	public void run() {
		Random random = new Random();
		// 模拟生成order_id(自增), order_amt, create_time, area_id
		String[] order_amt = { "10.10", "20.10", "50.2", "60.0", "80.1" };
		String[] area_id = { "1", "2", "3", "4", "5" };

		int i = 0;
		while (true) {
			i++;
			String messageStr = i + "\t" + order_amt[random.nextInt(5)] + "\t"
					+ DateFmt.getCountDate(null, DateFmt.date_long) + "\t"
					+ area_id[random.nextInt(5)];
			System.out.println("product: " + messageStr);
			producer.send(new KeyedMessage<Integer, String>(topic, messageStr));
			Utils.sleep(1000);
			// if (i == 10) {
			// break;
			// }
		}
	}

	public static void main(String[] args) {
		Producer producer = new Producer(KafkaProperties.Order_topic);
		producer.start();
	}
}
